package com.heima.kafka.sample;

import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * 消费者
 */
public class ConsumerQuickStart {

    public static void main(String[] args) {

        //1.kafka的配置信息(都可以写进application.yml里面)
        Properties prop = new Properties();
        //链接地址
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        //key和value的反序列化器
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        //设置消费者组(能够让只能一个消费者接收到消息)，想要都收到消息，就需要每个消费者都设置属于不同的组
        prop.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
        //修改成group2之后，哪个消费者运行了或者重新运行了，谁就属于哪一个组(右上角编辑配置，可以复制多个消费者)

        //手动提交偏移量
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);


        //2.创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop);

        //3.订阅主题//从哪个topic中拉取消息//subscribe()接收的是集合，发来的消息是个字符串，所以用Collections.singletonList()转换
        consumer.subscribe(Collections.singletonList("topic-first"));
        consumer.subscribe(Collections.singletonList("itcast-topic-out"));

        //4.拉取消息

        //同步和异步同时提交偏移量(异步提交失败之后，在用同步来保底)
        try {
            while (true) {//让它一直处于监听状态(不让当前线程死掉)                                  按照每1秒拉取一次的频率
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    System.out.println("拿到key了："+consumerRecord.key());
                    System.out.println("拿到value了："+consumerRecord.value());
                    //获取偏移量(是一个连续递增的数值，分区下的一个序列号，再去发送一条，就会 +1)
                    System.out.println("kafka没有消费者确认机制，他是根据偏移量offset记录到底是消费到什么位置了: "
                            + consumerRecord.offset());
                    //0号分区，0号虚拟机？？？
                    System.out.println("当前消息存储在哪一个分区下的分区号： "
                            + consumerRecord.partition());
                }
                //异步的方式提交偏移量
                consumer.commitAsync();
            }
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("记录错误的信息："+e);
        }finally {
            //异步产生错误之后，再用
            //同步方式提交偏移量
            consumer.commitSync();
        }


//        while (true){//让它一直处于监听状态(不让当前线程死掉)                                  按照每1秒拉取一次的频率
//            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
//            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//                System.out.println(consumerRecord.key());
//                System.out.println(consumerRecord.value());
//                System.out.println(consumerRecord.offset());
//                System.out.println(consumerRecord.partition());
//
//                try {
//                    //同步提交偏移量
//                    consumer.commitSync();
//                }catch (CommitFailedException e){
//                    System.out.println("记录提交失败的异常："+e);
//                }
//            }
//            //异步的方式提交偏移量
//                consumer.commitAsync(new OffsetCommitCallback() {
//                @Override
//                public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
//                    if(e != null){
//                        System.out.println("记录错误的提交偏移量："+map+",异常信息为："+e);
//                    }
//                }
//            });
//
//        }
    }

}
